/*
 * 文件名称：MsgClient.cpp
 * 文件描述：消息客户端主逻辑，用于与消息服务器之间传递消息
 * 作者信息：Satan-Emperor
 * 版本信息：V1.0.0
 * 修改历史：2014-04-22
 */

#include "MsgClient.h"
#include "MsgServerMessage.h"
#include "AppTimer.h"
#include "debugLog.h"
#include "SystemConfig.h"

#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/un.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <stdio.h>
#include <sys/epoll.h>
#include <errno.h>
#include <sys/select.h>
#include <netinet/tcp.h>
#include <signal.h>

/* 默认的消息服务器IP，如果采用普通的socket，则需要定义服务器IP */
#define DEFAULT_MSG_SERVER_IP		"127.0.0.1"

/* 消息服务器端口，由于采用AF_UNIX的高级进程间通信，故此不需要 */
#define DEFAULT_MSG_SERVER_PORT		1000

MsgClient::MsgClient(const char *msgServerIp, int port, int interval)
{
    if (msgServerIp != NULL)
    {
        strcpy( mMsgServerIp, msgServerIp );
    }
    else
    {
        strcpy( mMsgServerIp, DEFAULT_MSG_SERVER_IP );
    }

    if (port != 0)
    {
        mMsgServerPort = port;
    }
    else
    {
        mMsgServerPort = DEFAULT_MSG_SERVER_PORT;
    }

    mInterval = interval;

    mRecievedLen = 0;

    isThreadStopped = true;
    isThreadRun = false;

    msgTimeOutCnt = 0;
    lastDetectTime = 0;

    init();
}

MsgClient::~MsgClient()
{
    onDisconnect();

    close( mSocket );
    close( mEpoll );
}

int MsgClient::init()
{
    mSocket = socket(AF_UNIX, SOCK_STREAM, 0);

    linger lingerOpt;
    lingerOpt.l_onoff = 1;
    lingerOpt.l_linger = 0;
    setsockopt( mSocket, SOL_SOCKET, SO_LINGER, (char *)&lingerOpt, sizeof(linger) );

    int reuseAddr = 1;
    setsockopt( mSocket, SOL_SOCKET, SO_REUSEADDR, &reuseAddr, sizeof(reuseAddr) );

    int flag = 1;
    setsockopt( mSocket, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag) );

    struct sockaddr_un clientAddr;
    memset( &clientAddr, 0, sizeof(sockaddr_un) );
    clientAddr.sun_family = AF_UNIX;
    strcpy(clientAddr.sun_path, "/tmp/.MsgServerSocket");
    //clientAddr.sin_port = htons( mMsgServerPort );
    //inet_pton(AF_INET, mMsgServerIp, &clientAddr.sin_addr);
    if( connect( mSocket, ( struct sockaddr * ) &clientAddr, sizeof (sockaddr_un) ) == -1 )
    {
        DEBUG_API_ERROR(COMMON_LIB, "connect()");
        exit(-1);
    }

    fcntl( mSocket, F_SETFL, O_NONBLOCK );
    fcntl(mSocket, F_SETFD, 1);

    /*	struct epoll_event ev;
    	mEpoll = epoll_create( MAX_EPOLL_EVENT_CNT );
    	ev.data.fd = mSocket;
    	ev.events = EPOLLIN | EPOLLET | EPOLLHUP;
    	epoll_ctl( mEpoll, EPOLL_CTL_ADD, mSocket, &ev );*/

    isThreadRun = true;
    mMsgSendThread.createThread( 1, 1, 100 * 1024, msgSendThreadHandle, this );

    return 0;
}

int MsgClient::onRecieveMsg()
{
    int maxRetryCnt = 0;
    char recvBuf[MAX_MESSAGE_BUFF_SIZE];
    while (1)
    {
        int recvBytes = recv( mSocket, recvBuf, MAX_MESSAGE_BUFF_SIZE, 0 );

        if( recvBytes == 0 )
        {
            DEBUG_ERROR(COMMON_LIB, "Process:%d: client fd = %d connection closed!", getpid(), mSocket);
            return -1;
        }
        else if( recvBytes < 0 )
        {
            if( errno == EINTR )
                continue;
            else if( errno == EAGAIN )
                break;
            else
            {
                DEBUG_API_ERROR(COMMON_LIB, "recv()");
                return -1;
            }
        }
        else
        {
            msgHandle( recvBuf, recvBytes );
        }

        maxRetryCnt++;
        if( maxRetryCnt > 10 )
        {
            DEBUG_ERROR(COMMON_LIB, "Process:%d recv too many times, now sleep a while", getpid() );
            break;
        }

    }
    return 0;
}

void MsgClient::onDisconnect()
{
    isThreadRun = false;

    int retry = 0;
    while( !isThreadStopped )
    {
        if( retry++ > 100 )
            break;

        usleep( 10000 );
    }

    mMsgQueue.clearMsgQueue();

    /*	epoll_ctl( mEpoll, EPOLL_CTL_DEL, mSocket, NULL );
    	close( mSocket );*/
}

void MsgClient::waitMsg()
{
#if 1
    int ret = 0;
    fd_set fdRead, fdException;
    struct timeval tv;

    FD_ZERO( &fdRead );
    FD_ZERO( &fdException );

    FD_SET( mSocket, &fdRead );
    tv.tv_sec = 0;
    tv.tv_usec = mInterval * 1000;

    ret = select( 32, &fdRead, NULL, NULL, &tv );
    if( ret < 0 )
    {
        if( errno == EINTR )
            return;

        DEBUG_API_ERROR(COMMON_LIB, "select");
        usleep( 500000 ); // 500ms
        return;
    }
    else if( ret > 0 )
    {
        if( FD_ISSET(mSocket, &fdRead) )
        {
            if( onRecieveMsg() != 0 )
            {
                onDisconnect();
            }
        }

        if( FD_ISSET(mSocket, &fdException) )
        {
            onDisconnect();
        }
    }
    else
    {
        consumeMsg();
    }

#else
    struct epoll_event events[MAX_EPOLL_EVENT_CNT];
    int nfds = 0, i = 0;
    nfds = epoll_wait( mEpoll, events, MAX_EPOLL_EVENT_CNT, mInterval );
    for( i = 0; i < nfds; i++ )
    {
        if( events[i].data.fd == mSocket )
        {
            if( events[i].events & EPOLLIN ) //Recieved message from message server
            {
                if( onRecieveMsg() != 0 )
                {
                    onDisconnect();
                }
            }
            else if( events[i].events & EPOLLHUP ) //Client fd closed
            {
                onDisconnect();
            }
        }
    }
#endif
}

int MsgClient::registerMsgHandle(int msgId, MsgHandle handle)
{
    if( handle == NULL )
    {
        return -1;
    }

    mMsgMapTbl.erase( msgId ); //Delete message map defined previous

    mMsgMapTbl[msgId] = handle;

    MsgRegisterInfo msgRegisterInfo;
    msgRegisterInfo.msgId = msgId;
    msgRegisterInfo.timeOut = 500; //Temp

    sendMsg( MSG_SERVER_REGISTER_MSG, 0, (const char * ) &msgRegisterInfo, sizeof(MsgRegisterInfo), 0 , NULL);

    return 0;
}

int MsgClient::unRegisterMsgHandle(int msgId)
{
    mMsgMapTbl.erase( msgId );

    return 0;
}

int MsgClient::sendMsg(int msgId, char reply, const char *pData, int dataLen, int clientFd, int privData, char *jSonpFun)
{
    char *pMsgBuffer = NULL;
    pMsgBuffer = new char[dataLen + sizeof(MsgHead)];
    if( pMsgBuffer == NULL )
    {
        DEBUG_ERROR(COMMON_LIB, "Send message error, no enough memory!");
        return -1;
    }

    PMsgHead pMsgHead = ( PMsgHead ) pMsgBuffer;
    pMsgHead->msgId = msgId;
    pMsgHead->reply = reply;
    pMsgHead->dataLen = dataLen;
    pMsgHead->clientFd = clientFd;
    pMsgHead->privData = privData;
    pMsgHead->jSonpFun[0] = '\0';
    if( jSonpFun )
    {
        strncpy( pMsgHead->jSonpFun, jSonpFun, 20 );
    }
    if( (pData != NULL) && (dataLen != 0) )
    {
        memcpy( pMsgHead->data, pData, dataLen );
    }

    mMsgQueue.pushMsg( pMsgBuffer );

    return 0;
}

int MsgClient::replyMsg(PMsgHead pMsgHead, const char *pData, int dataLen)
{
    if( pMsgHead->reply == 1 )
    {
        return sendMsg( pMsgHead->msgId | REPLY_MESSAGE_MASK, 0, pData, dataLen, pMsgHead->clientFd, pMsgHead->privData, pMsgHead->jSonpFun);
    }
    else
    {
        return -1;
    }
}

int MsgClient::clearMsgQueue(void)
{
    mRecievedLen = 0;

    return 0;
}

void *MsgClient::msgSendThreadHandle(void *param)
{
    MsgClient *pThis = (MsgClient * )param;
    char *pMsgBuffer;
    int msgLen = 0;
    PMsgHead pMsgHead;
    int ret = -1;

    pThis->isThreadStopped = false;

    while( pThis->isThreadRun )
    {
        // Get message from queue
        ret = pThis->mMsgQueue.getMsg( &pMsgBuffer );
        if( ret == 0 ) // success
        {
            pMsgHead = ( PMsgHead ) pMsgBuffer;
            msgLen = pMsgHead->dataLen + sizeof(MsgHead);
            // Send message via socket
            if( pThis->sendDataToServer(pMsgBuffer, msgLen) != 0 )
            {
                DEBUG_ERROR(COMMON_LIB, "Process:%d send  msg:0x%x to msg server fail", getpid(), pMsgHead->msgId);
                delete []pMsgBuffer;
                pThis->mMsgQueue.clearMsgQueue();
                break; // The client will restart
            }
            else
            {
                // release message buffer memory
                delete []pMsgBuffer;
            }
        }
        else
        {
            usleep( 20000 );
        }
    }

    pThis->isThreadStopped = true;

    pThis->mMsgSendThread.exitThead();

    return NULL;
}

int MsgClient::sendDataToServer(const char *pData, int dataLen)
{
    int remainLen = dataLen;
    int sendLen = 0;
    fd_set fdSetWrite;
    struct timeval tv;
    const char *pSendPointer = pData;

    int retryCnt = 0;
    while( remainLen > 0 )
    {
        tv.tv_sec = 0;
        tv.tv_usec = 500000;

        FD_ZERO( &fdSetWrite );
        FD_SET( mSocket, &fdSetWrite );

        int ret = select( mSocket + 1, NULL, &fdSetWrite, NULL, &tv);
        if( ret > 0 ) //
        {
            sendLen = send( mSocket, pSendPointer, remainLen, 0 );
            if( sendLen  > 0 )
            {
                pSendPointer += sendLen;
                remainLen -= sendLen;
            }
            else if( sendLen == 0 ) //connection is close by peer
            {
                return -1;
            }
            else //sendLen < 0
            {
                if(errno == EINTR)
                    continue;
                else if(errno == EAGAIN)
                    continue;
                else
                    return -1;
            }
        }
        else if(ret < 0)
        {
            return -1;
        }

        retryCnt++;
        if( retryCnt > 20 ) //10s
        {
            DEBUG_ERROR(COMMON_LIB, "Process:%d has too many messages in send queue, now sleep a while", getpid() );
            break;
        }
    }

    return 0;
}

int MsgClient::msgHandle(const char *pRecvBuf, int bufLen)
{
    if( (pRecvBuf == NULL) || (bufLen <= 0) )
    {
        return -1;
    }

    if( (mRecievedLen + bufLen) > MAX_MESSAGE_BUFF_SIZE )
    {
        DEBUG_ERROR(COMMON_LIB, "Msg queue overflow, pid:%d", getpid() );
        return -1;
    }

    memcpy( mRecieveBuff + mRecievedLen, pRecvBuf, bufLen );
    mRecievedLen += bufLen;

    // Parse message data
    consumeMsg();

    return 0;
}

int MsgClient::consumeMsg(void)
{
    // Parse message data
    int parsedLen = 0;
    char *pParsePoint = NULL;
    PMsgHead pMsgHead = NULL;
    int msgDataLen = 0;
    int maxRetryCnt = 0;
    while( 1 )
    {
        /*
         * Message parse
         */
        pParsePoint = mRecieveBuff + parsedLen;

        int remainMsgDataLen = mRecievedLen - parsedLen;
        if( (unsigned int)remainMsgDataLen < sizeof(MsgHead) ) // Msg not complete
        {
            break;
        }

        pMsgHead = (PMsgHead )pParsePoint;

        if( (pMsgHead->dataLen + sizeof(MsgHead)) > (unsigned int)remainMsgDataLen ) // Msg not complete
        {
            break;
        }

        msgDataLen = (pMsgHead->dataLen + sizeof(MsgHead) );
        parsedLen += msgDataLen;

        /*
         * Message handle
         */
        if( mMsgMapTbl[pMsgHead->msgId] != NULL )
        {
            mMsgMapTbl[pMsgHead->msgId]( (const char * )pMsgHead, msgDataLen );
        }

        maxRetryCnt++;
        if( maxRetryCnt > 10 )
        {
            msgTimeOutCnt++;
            time_t timeTmp;
            time(&timeTmp);
            if (ABS(timeTmp - lastDetectTime) > 3 * 60)	// 3 min
            {
                if (msgTimeOutCnt >= 5)	// 5 times
                {
                    kill(getpid(), SIGTERM);
                }
                else
                {
                    msgTimeOutCnt = 0;
                    lastDetectTime = timeTmp;
                }
            }

            DEBUG_ERROR(MSG_SERVER, "Process:%d has too many messages in queue, now sleep a while", getpid() );
            break;
        }
    }

    /*
     * Message handle complete
     */
    if( mRecievedLen > 0 )
    {
        int remainDataLen = mRecievedLen - parsedLen;
        char *pRemainData = mRecieveBuff + parsedLen;
        memmove( mRecieveBuff, pRemainData, remainDataLen );
        mRecievedLen = remainDataLen;
    }

    return 0;
}

int MsgClient::msgCheck(const char *pRecvBuf, int bufLen)
{
    if( (unsigned int)bufLen < sizeof(MsgHead) )
    {
        DEBUG_ERROR(COMMON_LIB, "MsgClient::msgCheck error, recv too little data!");
        return -1;
    }

    PMsgHead pMsgHead = (PMsgHead ) pRecvBuf;
    if( ((pMsgHead->dataLen + sizeof(MsgHead) > (unsigned int)bufLen)) ||
            (pMsgHead->dataLen > MAX_MESSAGE_BUFF_SIZE) )
    {
        return -1;
    }

    return 0;
}

//EOF
